package com.hong.spring.listener;

import com.alibaba.fastjson.JSONObject;
import com.hong.spring.api.IUserService;
import com.hong.spring.entity.User;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.listener.MessageListener;

/**
 * @author: csh
 * @Date: 2021/3/16 11:14
 * @Description:用户监听
 */
@Log4j2
public class UserListener implements MessageListener<String,String> {

    @Autowired
    private IUserService userService;


    @Override
    public void onMessage(ConsumerRecord<String, String> record) {
        try {
            log.info(String.format("消费者1收到了消息。topic: %s, partition: %s, offset: %s, key: %s, value: %s", record.topic(),
                    record.partition(), record.offset(), record.key(), record.value()));
            User user = JSONObject.parseObject(record.value(), User.class);
            log.info("获取的用户信息{}", JSONObject.toJSONString(user));
            if(null!=user){
                userService.save(user);
            }
        }catch (Exception e){
            log.error("添加用户异常{}",e);
        }
    }
}
